package ws

import (
	"context"
)

type MessageManager struct {
	writeQueue     chan *MessageJob
	readQueue      chan *MessageJob
	sessionManager *SessionManager
	sessionBuilder *SessionBuilder
	ctx            context.Context
}

func NewMessageManager(sessionManager *SessionManager, sessionBuilder *SessionBuilder, ctx context.Context) *MessageManager {
	return &MessageManager{
		sessionManager: sessionManager,
		ctx:            ctx,
		sessionBuilder: sessionBuilder,
	}
}

func (x *MessageManager) init() {
	x.writeQueue = make(chan *MessageJob, x.sessionBuilder.WriteQueueSize)
	x.readQueue = make(chan *MessageJob, x.sessionBuilder.ReadQueueSize)
	go x.read()
	go x.write()
}

func (x *MessageManager) Write(session *Session, wsPackage *WsPackage) {
	x.writeQueue <- NewMessageJob(session, NewWsPackage(wsPackage.MessageType, wsPackage.Data))
}

func (x *MessageManager) Read(session *Session, wsPackage *WsPackage) {
	x.readQueue <- NewMessageJob(session, NewWsPackage(wsPackage.MessageType, wsPackage.Data))
}

func (x *MessageManager) write() {
	for msg := range x.writeQueue {
		msg.session.write(msg.wsPackage)
	}
}

func (x *MessageManager) read() {
	for msg := range x.readQueue {
		msg.session.read(msg.wsPackage)
	}
}

func (x *MessageManager) GroupSend(wsPackage *WsPackage, filter func(session *Session) bool) {
	x.sessionManager.IterateSession(func(session *Session) {
		if filter == nil || filter(session) {
			x.Write(session, wsPackage)
		}
	})
}

func (x *MessageManager) Broadcast(wsPackage *WsPackage) {
	x.GroupSend(wsPackage, nil)
}

func (x *MessageManager) Close() {
	close(x.readQueue)
	close(x.writeQueue)
}
